package fun.hijklmn.magic.resource.server.service.impl;

import com.alibaba.fastjson.JSONObject;
import fun.hijklmn.magic.resource.common.enums.FileTypeEnum;
import fun.hijklmn.magic.resource.server.config.ResourceProperties;
import fun.hijklmn.magic.resource.server.service.MQService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;


/**
 * @Description
 * @Author fs
 * @Date 2021/6/2 20:49
 * @Version 0.0.1-SNAPSHOT
 **/
@Slf4j
@Service
public class MQServiceImpl implements MQService {

    @Autowired
    private ResourceProperties resourceProperties;

    @Autowired
    private KafkaTemplate<Object, Object> kafkaTemplate;

    @Override
    @Async
    public void send(String message) {

        try {

            JSONObject jsonObject = new JSONObject();

            jsonObject.put("url", message);
            jsonObject.put("type", "count");
            jsonObject.put("resType", "other");

            if (message.contains(".")) {
                String type = message.substring(message.lastIndexOf(".") + 1);
                if (!CollectionUtils.isEmpty(resourceProperties.getImageCondition()) && resourceProperties.getImageCondition().contains(type)) {
                    jsonObject.put("resType", FileTypeEnum.IMAGE);
                } else if (!CollectionUtils.isEmpty(resourceProperties.getDocumentCondition()) && resourceProperties.getDocumentCondition().contains(type)) {
                    jsonObject.put("resType", FileTypeEnum.DOCUMENT);
                } else if (!CollectionUtils.isEmpty(resourceProperties.getSoundCondition()) && resourceProperties.getSoundCondition().contains(type)) {
                    jsonObject.put("resType", FileTypeEnum.SOUND);
                } else if (!CollectionUtils.isEmpty(resourceProperties.getVideoCondition()) && resourceProperties.getVideoCondition().contains(type)) {
                    jsonObject.put("resType", FileTypeEnum.VIDEO);
                }
            }

            log.info("----------->发送分析数据，json=" + jsonObject.toJSONString());

            kafkaTemplate.send(resourceProperties.getTopic(), jsonObject.toJSONString());

        } catch (Exception e) {
            log.error("------------->发送统计信息失败,message=" + message, e);
        }

    }

}
